﻿using Confluent.Kafka;
using Kafka.TeacherApi.Model;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Caching.Distributed;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

namespace Kafka.TeacherApi.Controllers
{
    public class TeacherController : ControllerBase
    {
        private readonly ILogger<TeacherController> _logger;
        private readonly IDistributedCache _distributedCache;

        public TeacherController(ILogger<TeacherController> logger, IDistributedCache distributedCache)
        {
            _logger = logger;
            _distributedCache = distributedCache;
        }

        /// <summary>
        ///  自动确认-不会重复消费
        /// </summary>
        /// <returns></returns>
        [HttpPost("noticeCreate1")]
        public string HandleNotice1()
        {
            new Task(() =>
            {
                var consumerConfig = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    GroupId = "notices",
                    EnableAutoCommit = true //消息自动确认、防止重复消费，但是不会管收到消息后的业务逻辑是否处理成功，
                };
                var builder = new ConsumerBuilder<string, string>(consumerConfig);

                using (var consumer = builder.Build())
                {
                    // 1、订阅
                    consumer.Subscribe("noticeCreate1");
                    while (true) //循环监听消费
                    {
                        try
                        {
                            // 2、消费(自动确认)
                            var result = consumer.Consume();

                            // 3、业务逻辑:业务逻辑---->执行失败--->消息丢失
                            string key = result.Key;
                            string value = result.Value;

                            _logger.LogInformation($"Teacher接收通知：Key:{key}、执行任务,Value:{value}");
                        }
                        catch (Exception e)
                        {
                            _logger.LogInformation($"异常：{e}");
                        }
                    }
                }
            }).Start();

            Console.WriteLine("开始监听...");
            return "AutoOffsetReset.Earliest=》开启监听=》noticeCreate1";
        }

        /// <summary>
        /// 手动确认2
        /// </summary>
        /// <returns></returns>
        [HttpPost("noticeCreate2")]
        public ResultObject HandleNotice2()
        {
            /*
             * 自动确认的缺陷：如果业务逻辑代码执行失败，该消息已确认后不会重复消费，则可能导致业务逻辑无法再重复执行。
             * 真实项目场景中应该使用手动确认消息，除非需要执行的业务即使丢失也没有关系
             * Rabbitmq消费完一个消息后会自动删除，Kafka可以永久存储，Kafka根据偏移量来读取消息，根据分区存储消费者消息，可以多次尝试
             * Kafka没有交换机，主要是根据GroupId（类似RabbitMQ的基本交换机的概念）来区分
             * 消息量不大的情况适用RabbitMQ
            
             */
            new Task(() =>
            {
                var consumerConfig = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    GroupId = "notices2",
                    EnableAutoCommit = false, //配置为手动确认，可以获取消费消息的偏移量，确保消费后的业务逻辑正常。
                };
                var builder = new ConsumerBuilder<string, string>(consumerConfig);
                using (var consumer = builder.Build())
                {
                    // 1、订阅
                    consumer.Subscribe("noticeCreate2");
                    // 1.1、重置偏移量
                    consumer.Assign(new TopicPartitionOffset(new TopicPartition("noticeCreate2", 0), 0));
                    while (true)
                    {
                        // 2、消费
                        var result = consumer.Consume();

                        // 2.1、获取偏移量（偏移量就是Kafka集合中的序号）
                        _logger.LogInformation($"消息偏移量：Offset:{result.Offset}");

                        // 3、业务处理
                        string key = result.Key;
                        string value = result.Value;
                        _logger.LogInformation($"消息：Key:{key}");
                        _logger.LogInformation($"消息：Message:{value}");

                        // 3、手动提交（向kafka确认消息）
                        //consumer.Commit(result);

                        // 手动提交、效率比consumer.Commit效率更高，不要同时使用2种提交方式
                        consumer.StoreOffset(result);
                    }
                }
            }).Start();

            Console.WriteLine("监听通知...");

            return new ResultObject
            {
                message = "noticeCreate2-监听通知...",
                code = 200,
            };
        }

        /// <summary>
        /// 手动确认(存储偏移量) noticeCreate2
        ///  * Kafka防止重复消费，可以使用Redis存储偏移量（项目宕机，Kafka提交失败，需要存储偏移量）
        /// </summary>
        /// <returns></returns>
        [HttpPost("noticeCreate2-2")]
        public ResultObject HandleNotice3()
        {
            new Task(() =>
            {
                var consumerConfig = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    GroupId = "notices",
                    EnableAutoCommit = false,
                };
                var builder = new ConsumerBuilder<string, string>(consumerConfig);
                using (var consumer = builder.Build())
                {
                    // 1、订阅
                    consumer.Subscribe("noticeCreate2");

                    // 1.2、获取偏移量
                    string offset = _distributedCache.GetString("noticeCreate2");
                    if (string.IsNullOrEmpty(offset))
                    {
                        offset = "0";
                    }

                    // 1.3、重置偏移量
                    consumer.Assign(new TopicPartitionOffset(new TopicPartition("noticeCreate2", 0), int.Parse(offset) + 1));
                    while (true)
                    {
                        // 2、消费
                        var result = consumer.Consume();

                        // 2.1、获取偏移量
                        _logger.LogInformation($"订单消息偏移量：Offset:{result.Offset}");
                        // 2.2、把kafka队列中偏移量存起来。redis mysql
                        // 2.3、重置kafka队列的偏移量，存储偏移量
                        _distributedCache.SetString("noticeCreate2", result.Offset.Value.ToString());

                        // 3、业务处理
                        string key = result.Key;
                        string value = result.Value;
                        _logger.LogInformation($"Key:{key}");
                        _logger.LogInformation($"Order:{value}");

                        // redis缺陷：无法保证偏移和业务同时成功。
                        // TODO 方案：使用数据库来存储偏移量
                        //       核心：通过数据库【事务】来保证，偏移量和对应操作业务要关联起来
                        // 3、手动提交
                        // consumer.Commit(result);
                    }
                }
            }).Start();

            Console.WriteLine("创建监听......");
            return new ResultObject
            {
                message = "创建监听......",
                code = 200
            };
        }

        /// <summary>
        /// 创建通知集群-消息分区-固定分区 noticeCreate2
        /// Kafka根据主题+分区存储对应的偏移量，宕机时Kafka会自动存储偏移量到磁盘。
        /// </summary>
        /// <returns></returns>
        [HttpPost("noticeCreate2-3")]
        public ResultObject HandleNotice4()
        {

            new Task(() =>
            {
                var consumerConfig = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    GroupId = "order", //针对多个不同消费者，使用相同的Topic不同的GroupId来分组，
                    EnableAutoCommit = false,
                };
                var builder = new ConsumerBuilder<string, string>(consumerConfig);
                var consumer = builder.Build();

                // 1、订阅
                consumer.Subscribe("noticeCreate2");
                // 1.1、获取偏移量
                string offset = _distributedCache.GetString("noticeCreate2");
                if (string.IsNullOrEmpty(offset))
                {
                    offset = "0";
                }
                // 1.2、重置偏移量
                consumer.Assign(new TopicPartitionOffset(new TopicPartition("noticeCreate2", 0), int.Parse(offset)));
                while (true)
                {
                    // 2、消费
                    var result = consumer.Consume();
                    // 2.1、获取偏移量
                    _logger.LogInformation($"偏移量：Offset:{result.Offset}");
                    _logger.LogInformation($"分区序号：Partition:{result.Partition}");

                    // 3、业务处理
                    string key = result.Key;
                    string value = result.Value;
                    _logger.LogInformation($"Key:{key}");
                    _logger.LogInformation($"Order:{value}");

                    // 2.2、把kafka队列中偏移量存起来。redis mysql
                    // 2.3、重置kafka队列的偏移量
                    _distributedCache.SetString("noticeCreate2", result.Offset.Value.ToString());

                    // 3、手动提交
                    consumer.Commit(result);
                }
            }).Start();

            Console.WriteLine("创建监听......");
            return new ResultObject
            {
                message = "创建监听......",
                code = 200,
            };
        }

        /// <summary>
        /// noticesDirect
        /// </summary>
        /// <returns></returns>
        [HttpPost("noticesDirect")]
        public ResultObject HandleNotice5()
        {
            new Task(() =>
            {
                var consumerConfig = new ConsumerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    GroupId = "notices",
                    EnableAutoCommit = false,
                    FetchMinBytes = 170,
                    FetchMaxBytes = 3060
                };
                var builder = new ConsumerBuilder<string, string>(consumerConfig);
                using (var consumer = builder.Build())
                {
                    // 1、订阅
                    consumer.Subscribe("noticesDirect");
                    // 2、偏移量恢复
                    string offset = _distributedCache.GetString("noticesDirect");
                    if (string.IsNullOrEmpty(offset))
                    {
                        offset = "0";
                    }
                    consumer.Assign(new TopicPartitionOffset(new TopicPartition("noticesDirect", 0), int.Parse(offset)));
                    while (true)
                    {
                        // 1、恢复消息
                        new Timer((s) =>
                        {
                            consumer.Resume(new List<TopicPartition> { new TopicPartition("noticesDirect", 0) });
                        }, null, Timeout.Infinite, Timeout.Infinite).Change(5000, 5000);

                        // 1.1、消费暂停
                        consumer.Pause(new List<TopicPartition> { new TopicPartition("noticesDirect", 0) });

                        // 2、消费消息
                        var result = consumer.Consume(); //批量获取消息，根据-----》字节数
                        try
                        {
                            // 2.1、获取偏移量
                            _logger.LogInformation($"偏移量：Offset:{result.Offset}");

                            // 3、业务处理
                            string key = result.Key;
                            string value = result.Value;
                            _logger.LogInformation($"创建商品：Key:{key}");
                            _logger.LogInformation($"创建商品：Order:{value}");

                            // 2.2、把kafka队列中偏移量存起来。redis mysql
                            // 2.3、重置kafka队列的偏移量
                            _distributedCache.SetString("noticesDirect", result.Offset.Value.ToString());

                            // 3、手动提交
                            consumer.Commit(result);
                        }
                        catch (Exception)
                        {

                            throw;
                        }
                        finally
                        {
                            consumer.Pause(new List<TopicPartition> { new TopicPartition("noticesDirect", 0) });
                            Console.WriteLine($"暂停消费");
                        }
                    }
                }

            }).Start();
            Console.WriteLine("创建监听......");
            return new ResultObject
            {
                message = "noticesDirect 创建监听......",
                code = 200
            };

        }

        /// <summary>
        /// 监听广播（主题交换机-手动确认）
        /// </summary>
        /// <returns></returns>
        [HttpPost("broadcast1")]
        public ResultObject HandleBroadcast1()
        {
            /*
             *  自动确认机制缺陷：
             *  1、消息是否正常添加到数据库当中,所以需要使用手工确认
             * **/

            //1、创建链接
            var factory = new ConnectionFactory()
            {

                HostName = "localhost",
                Port = 5672,
                UserName = "guest",
                Password = "guest",
                VirtualHost = "/"
            };
            var connection = factory.CreateConnection();
            var channel = connection.CreateModel();

            //2、定义交换机
            channel.ExchangeDeclare(exchange: "teacher_topic", type: ExchangeType.Topic);

            //3、定义队列（随机）
            var queueName = channel.QueueDeclare().QueueName;

            //4、队列与交换机绑定（多对一）
            // * 号的缺陷：只能匹配一级
            // # 能够匹配一级及多级以上
            // 真实项目当中，使用主题交换机。因为：可以满足所有场景
            channel.QueueBind(queueName, "teacher_topic", routingKey: "sms.#");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {

                var body = ea.Body;
                //4.1:业务逻辑代码...
                //...
                var message = Encoding.UTF8.GetString(body.ToArray());
                Console.WriteLine($"收到消息...{message}");
                //确认消息
                channel.BasicAck(ea.DeliveryTag, true);
            };

            //5、消费消息 每次一个消费者只成功消费一个消息
            channel.BasicQos(0, 1, false);

            channel.BasicConsume(queue: queueName,
                autoAck: false,//消息确认、防止消费失败
                consumer: consumer);
            Console.WriteLine("创建消息监听...");
            return new ResultObject
            {
                code = 200,
                message = "消息监听成功..."
            };
        }
    }
}
